package com.ruyuan.event.streaming.process;
import com.ruyuan.event.streaming.pojo.*;
import com.ruyuan.event.streaming.utils.Constants;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.Jedis;

import java.io.IOException;

public class ProcessETLUtils {

    public static final String RETRY_PARTITIONS = Constants.KAFKA_EVENT_RETRY_LOG;
    static protected final Logger log = LoggerFactory.getLogger(ProcessETLUtils.class);


    /**
     * 生成redis的存储key
     * */

    public static byte[] generateBytesKey(EventClientLog eventClientLog) {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(eventClientLog.getRequestId());
        stringBuilder.append(eventClientLog.getEventId());
        stringBuilder.append(eventClientLog.getEventUserId());
        return stringBuilder.toString().getBytes();
    }

    public static byte[] generateBytesKey(EventJoinLog eventJoinLog) {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(eventJoinLog.getRequestId());
        stringBuilder.append(eventJoinLog.getEventId());
        stringBuilder.append(eventJoinLog.getEventUserId());
        return stringBuilder.toString().getBytes();
    }

    public static byte[] generateBytesKey(EventServerLog eventServerLog) {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(eventServerLog.getRequestId());
        stringBuilder.append(eventServerLog.getEventId());
        stringBuilder.append(eventServerLog.getEventUserId());
        return stringBuilder.toString().getBytes();
    }

    public static EventServerLog generateServerContext(EventServerLog eventServerLog) {
        EventServerLog.Builder builder = EventServerLog.newBuilder();
        /**
         * 设置需要保存到缓存的数据
         *
         * //客户端的标识信息
         *     string imei = 21;
         *     string ip = 22;
         *     string network = 23;
         *
         *     //根据当前用户的标签数据匹配对应的活动信息
         *     string user_id = 31;
         *     string locale = 32;
         *     uint64 birthyear = 33;
         *     string gender = 34;
         *     uint64 joinedAt = 35;
         *     string location = 36;
         *     string country = 37;
         *     string province = 38;
         *     string city = 39;
         *     string timezone = 40;
         * */
        builder.setRequestId(eventServerLog.getRequestId());
        builder.setTimestamp(eventServerLog.getTimestamp());

        builder.setLocale(eventServerLog.getLocale());
        builder.setBirthyear(eventServerLog.getBirthyear());
        builder.setGender(eventServerLog.getGender());
        builder.setJoinedAt(eventServerLog.getJoinedAt());
        builder.setLocation(eventServerLog.getLocation());
        builder.setCountry(eventServerLog.getCountry());
        builder.setProvince(eventServerLog.getProvince());
        builder.setCity(eventServerLog.getCity());
        builder.setTimezone(eventServerLog.getTimezone());
        return builder.build();
    }

    public static void writeRedis(Jedis jedis, byte[] key, EventServerLog eventSerLog) {
        jedis.set(key,eventSerLog.toByteArray());
        jedis.expire(key, Constants.DEFAULT_TTL);
    }

    public static void writeHbase(HTable hTable,byte[] key,EventServerLog eventServerLog) {
        log.info("servlerLog写入Hbase，eventServerLog： {}",eventServerLog.toString());
        try {
            Put put = new Put(key);
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_BASIC,"request_id".getBytes(), Bytes.toBytes(eventServerLog.getRequestId()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_BASIC,"timestamp".getBytes(), Bytes.toBytes(eventServerLog.getTimestamp()));
            //userInfo
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"imei".getBytes(), Bytes.toBytes(eventServerLog.getImei()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"ip".getBytes(), Bytes.toBytes(eventServerLog.getIp()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"network".getBytes(), Bytes.toBytes(eventServerLog.getNetwork()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"user_id".getBytes(), Bytes.toBytes(eventServerLog.getUserId()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"locale".getBytes(), Bytes.toBytes(eventServerLog.getLocale()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"birthyear".getBytes(), Bytes.toBytes(eventServerLog.getBirthyear()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"gender".getBytes(), Bytes.toBytes(eventServerLog.getGender()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"joinedAt".getBytes(), Bytes.toBytes(eventServerLog.getJoinedAt()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"location".getBytes(), Bytes.toBytes(eventServerLog.getLocation()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"country".getBytes(), Bytes.toBytes(eventServerLog.getCountry()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"province".getBytes(), Bytes.toBytes(eventServerLog.getProvince()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"city".getBytes(), Bytes.toBytes(eventServerLog.getCity()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_USERINFO,"timezone".getBytes(), Bytes.toBytes(eventServerLog.getTimezone()));
            //活动相关的数据
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_EVENTINFO,"event_id".getBytes(), Bytes.toBytes(eventServerLog.getEventId()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_EVENTINFO,"event_user_id".getBytes(), Bytes.toBytes(eventServerLog.getEventUserId()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_EVENTINFO,"event_start_time".getBytes(), Bytes.toBytes(eventServerLog.getEventStartTime()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_EVENTINFO,"event_city".getBytes(), Bytes.toBytes(eventServerLog.getEventCity()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_EVENTINFO,"event_country".getBytes(), Bytes.toBytes(eventServerLog.getEventCountry()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_EVENTINFO,"event_province".getBytes(), Bytes.toBytes(eventServerLog.getEventProvince()));
            put.addColumn(Constants.HBASE_EVENT_SERVER_COLUMN_FAMILY_EVENTINFO,"event_type".getBytes(), Bytes.toBytes(eventServerLog.getEventType()));
            hTable.put(put);
        }catch (IOException e) {
            e.printStackTrace();
        }
    }

    //读取redis或者HBase内容
    public static EventServerLog getServerContext(Jedis jedis, HTable hTable, byte[] key) {
        EventServerLog context = getContext(jedis,key);
        //缓存没有命中
        if (context == null) {
            context = getContext(hTable,key);
        }
        return context;
    }

    public static EventServerLog getContext(Jedis jedis, byte[] key) {
        byte[] data = jedis.get(key);
        if (data == null) {
            return null;
        }else {
            try {
                return EventServerLog.parseFrom(data);
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    private static EventServerLog getContext(HTable hTable, byte[] key) {
        Get get = new Get(key);
        try {
            Result result = hTable.get(get);
            if (result.getRow() == null) return null;
            return EventServerLog.parseFrom(result.getRow());
        }catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    public static EventJoinLog buildEventServerLog(EventClientLog eventClientLog, EventServerLog eventServerContext) {
        //构建EventJoinLog
        EventJoinLog.Builder eventJoinLogBuilder = EventJoinLog.newBuilder();
        ProcessInfo.Builder processInfoBuilder = ProcessInfo.newBuilder();
        processInfoBuilder.setProcessTimestamp(System.currentTimeMillis());

        //参数基本封装
        eventJoinLogBuilder.setRequestId(eventClientLog.getRequestId());
        eventJoinLogBuilder.setTimestamp(eventClientLog.getTimestamp());

        /**
         *      拼接客户端信息
         *      //客户端的标识信息
         *     string imei = 11;
         *     string ip = 12;
         *     string network = 13;
         *     string net_type = 14;
         *     string os_name = 15;
         *     string os_version = 16;
         *     string user_id = 31;
         * */
        eventJoinLogBuilder.setUserId(eventClientLog.getUserId());
        eventJoinLogBuilder.setImei(eventClientLog.getImei());
        eventJoinLogBuilder.setIp(eventClientLog.getIp());
        eventJoinLogBuilder.setNetwork(eventClientLog.getNetwork());
        eventJoinLogBuilder.setNetType(eventClientLog.getNetType());
        eventJoinLogBuilder.setOsName(eventClientLog.getOsName());
        eventJoinLogBuilder.setOsVersion(eventClientLog.getOsVersion());

        //满足拼接条件
        if (eventServerContext != null) {
            //拼接成功
            processInfoBuilder.setJoinServerLog(true);
            //拼接服务端信息
            joinServerLogContext(eventJoinLogBuilder,eventServerContext);
        }else {
            //不满足拼接条件
            processInfoBuilder.setRetryCount(1);
            processInfoBuilder.setJoinServerLog(false);
        }

        /**
         * 拼接关联活动信息
         *      由用户关联活动，所以拼接的是client端的活动信息
         * string event_id = 50;
         *     //活动举办者
         *     string event_user_id = 51;
         *     //活动开始时间
         *     uint64 event_start_time = 52;
         *     //活动举办的城市
         *     string event_city = 53;
         *     //活动举办的国家
         *     string event_country = 54;
         *     //活动举办的省份
         *     string event_province = 55;
         *     //活动的参与情况：报名，不报名，报名后参加，报名后不参加，不报名后参加活动
         *     string event_type = 60;
         *
         * */
        eventJoinLogBuilder.setEventId(eventClientLog.getEventId());
        eventJoinLogBuilder.setEventUserId(eventClientLog.getUserId());
        eventJoinLogBuilder.setEventStartTime(eventClientLog.getEventStartTime());
        eventJoinLogBuilder.setEventCity(eventClientLog.getEventCity());
        eventJoinLogBuilder.setEventCountry(eventClientLog.getEventCountry());
        eventJoinLogBuilder.setEventProvince(eventClientLog.getEventProvince());
        eventJoinLogBuilder.setEventType(eventClientLog.getEventType());

        switch (eventClientLog.getEventType()) {
            case "JOIN":
                eventJoinLogBuilder.setJoin(1);
                break;
            case "NOT_JOIN":
                eventJoinLogBuilder.setNotJoin(1);
                break;
            case "INVITED_JOIN":
                eventJoinLogBuilder.setInvitedJoin(1);
                break;
            case "NOT_INVITED_JOIN":
                eventJoinLogBuilder.setNotInvitedJoin(1);
                break;
            case "INTEREST":
                eventJoinLogBuilder.setInterest(1);
                break;
            case "NOT_INTEREST":
                eventJoinLogBuilder.setNotInterest(1);
                break;
            default:
                break;
        }
        return eventJoinLogBuilder.build();
    }

    //拼接serverLog到EventJoinLog
    public static void joinServerLogContext(EventJoinLog.Builder eventJoinLogBuilder, EventServerLog eventServerContext) {
        /**
         *     string locale = 32;
         *     uint64 birthyear = 33;
         *     string gender = 34;
         *     uint64 joinedAt = 35;
         *     string location = 36;
         *     string country = 37;
         *     string province = 38;
         *     string city = 39;
         *     string timezone = 40;
         * */
        eventJoinLogBuilder.setLocale(eventServerContext.getLocale());
        eventJoinLogBuilder.setBirthyear(eventServerContext.getBirthyear());
        eventJoinLogBuilder.setGender(eventServerContext.getGender());
        eventJoinLogBuilder.setJoinedAt(eventServerContext.getJoinedAt());
        eventJoinLogBuilder.setLocation(eventServerContext.getLocation());
        eventJoinLogBuilder.setCountry(eventServerContext.getCountry());
        eventJoinLogBuilder.setProvince(eventServerContext.getProvince());
        eventJoinLogBuilder.setCity(eventServerContext.getCity());
        eventJoinLogBuilder.setTimezone(eventServerContext.getTimezone());
        eventJoinLogBuilder.setTimestamp(eventServerContext.getTimestamp());
    }

    /**
     * 拼接失败的情况写入重试topic
     * */
    public static void retry(Producer producer, byte[] data) {
        sendRetryPartitions(producer,RETRY_PARTITIONS,data);
    }

    //重试数据写入
    private static void sendRetryPartitions(Producer producer, String topic, byte[] data) {
        producer.send(new ProducerRecord<String,byte[]>(topic,data));
    }

    public static void sendKafkaTopic(Producer producer, String topic, byte[] data) {
        ProducerRecord<String, byte[]> stringProducerRecord = new ProducerRecord<>(topic, data);
        producer.send(stringProducerRecord);
    }

    //完成DTO参数封装
    public static EventLogDTO buidEventLogDTO(EventJoinLog eventJoinLog) {
        EventLogDTO eventLogDTO = new EventLogDTO();
        eventLogDTO.setRequestId(eventJoinLog.getRequestId());
        eventLogDTO.setTimestamp(eventJoinLog.getTimestamp());
        eventLogDTO.setImei(eventJoinLog.getImei());
        eventLogDTO.setIp(eventJoinLog.getIp());
        eventLogDTO.setNetwork(eventJoinLog.getNetwork());
        eventLogDTO.setNetType(eventJoinLog.getNetType());
        eventLogDTO.setOsName(eventJoinLog.getOsName());
        eventLogDTO.setOsVersion(eventJoinLog.getOsVersion());
        eventLogDTO.setUserId(eventJoinLog.getUserId());
        eventLogDTO.setLocale(eventJoinLog.getLocale());
        eventLogDTO.setBirthyear(eventJoinLog.getBirthyear());
        eventLogDTO.setGender(eventJoinLog.getGender());
        eventLogDTO.setJoinedAt(eventJoinLog.getJoinedAt());
        eventLogDTO.setLocation(eventJoinLog.getLocation());
        eventLogDTO.setCountry(eventJoinLog.getCountry());
        eventLogDTO.setCity(eventJoinLog.getCity());
        eventLogDTO.setProvince(eventJoinLog.getProvince());
        eventLogDTO.setTimezone(eventJoinLog.getTimezone());
        eventLogDTO.setEventId(eventJoinLog.getEventId());
        eventLogDTO.setEventUserId(eventJoinLog.getEventUserId());
        eventLogDTO.setEventStartTime(eventJoinLog.getEventStartTime());
        eventLogDTO.setEventCity(eventJoinLog.getEventCity());
        eventLogDTO.setEventCountry(eventJoinLog.getEventCountry());
        eventLogDTO.setEventProvince(eventJoinLog.getEventProvince());
        //设置状态
        eventLogDTO.setEventType(eventJoinLog.getEventType());
        eventLogDTO.setJoin(eventJoinLog.getJoin());
        eventLogDTO.setNotJoin(eventJoinLog.getNotJoin());
        eventLogDTO.setInvitedJoin(eventJoinLog.getInvitedJoin());
        eventLogDTO.setNotInvitedJoin(eventJoinLog.getNotInvitedJoin());
        eventLogDTO.setInterest(eventJoinLog.getInterest());
        eventLogDTO.setNotInterest(eventJoinLog.getNotInterest());
        return eventLogDTO;
    }
}
